// Database datapump implementation
// Copyright 2006 (C) Ralph Thomas

#include <datapumps/database.h>
#include <binder.hpp>
#include <util/db.h>
#include <util/resources.h>
#include <adobe/array.hpp>
#include <boost/bind.hpp>
#include <ext/hash_fun.h>
#include <util/concepts/list.h>

#include <iostream>
#include <sstream>

namespace bfs = boost::filesystem;
namespace sgi = __gnu_cxx;

namespace datapumps {
	typedef boost::shared_ptr<adobe::dictionary_t> dictionary_ptr_t;
	
	
	struct db_row_value_t {
		db_row_value_t() : row_number_m(0), generation_m(-1) {}
	
		db_row_value_t(dictionary_ptr_t rows, unsigned int i, const std::string& db_name, int generation)
		 : rows_m(rows), row_number_m(i), db_name_m(db_name), generation_m(generation) {}
		
		adobe::value_t get_column(const adobe::name_t& column) {
			if( !rows_m ) throw std::runtime_error( "db_row_value_t::get_column called with null result set" );
			if( !rows_m->count(column) ) return adobe::value_t();
			util::concept::List l( util::concept::get_list(rows_m->at(column)) );
			if( l.empty() || l.size() <= row_number_m ) return adobe::value_t();
			return l.at( row_number_m );
		}
		
		adobe::value_t get_entire_column(const adobe::name_t& column) {
			if( !rows_m ) throw std::runtime_error( "db_row_value_t::get_column called with null result set" );
			if( !rows_m->count(column) ) return adobe::value_t();
			return rows_m->at(column);
		}
		
		adobe::dictionary_t get_row_dictionary() {
			if( !rows_m ) throw std::runtime_error( "db_row_value_t::get_row_dictionary called with null result set" );
			adobe::dictionary_t out;
			for( adobe::dictionary_t::const_iterator i = rows_m->begin(); i != rows_m->end(); i++ ) {
				util::concept::List l( util::concept::get_list( i->second ) );
				adobe::value_t v( l.at( row_number_m ) );
				if (v.type() == typeid(double)) {
					std::stringstream ss; ss << v.get<double>();
					v = adobe::value_t(ss.str());	
				}
				out.insert(adobe::dictionary_t::value_type(i->first, v));
			}
			return out;
		}
		
		bool operator==( const db_row_value_t& rhs ) const {
			if( generation_m == rhs.generation_m )
				return (row_number_m == rhs.row_number_m);
			//
			// Okay, these are from different generations. We'll have to
			// actually compare the values :(.
			//
			if ((rows_m == 0) && (rhs.rows_m == 0)) return true;
			if ((rows_m == 0) || (rhs.rows_m == 0)) return false;
			if (rows_m->size() != rhs.rows_m->size()) return false;
			
			adobe::dictionary_t::const_iterator us = rows_m->begin();
			adobe::dictionary_t::const_iterator them = rhs.rows_m->begin();
			adobe::dictionary_t::const_iterator us_end = rows_m->end();
			adobe::dictionary_t::const_iterator them_end = rhs.rows_m->end();
			while( us != us_end ) {
				if( us->first != them->first ) return false;
				util::concept::List us_list = util::concept::get_list( us->second );
				util::concept::List them_list = util::concept::get_list( them->second );
				if( (us_list.size() <= row_number_m) && (them_list.size() <= rhs.row_number_m) )
				{
					us++;
					them++;
					continue;
				}
				if( (us_list.size() <= row_number_m) || (them_list.size() <= rhs.row_number_m) )
					return false;
				if( us_list.at(row_number_m) != them_list.at(rhs.row_number_m) )
					return false;
				us++;
				them++;
			}
			return true;
		}
		
		db_row_value_t& operator=( const db_row_value_t& rhs ) {
			rows_m = rhs.rows_m;
			row_number_m = rhs.row_number_m;
			db_name_m = rhs.db_name_m;
			return *this;
		}
		
		dictionary_ptr_t			rows_m;
		unsigned int				row_number_m;
		std::string					db_name_m;
		int							generation_m;
	};

	struct db_list_t : public util::concept::ListInterface {
		std::size_t					length_m;
		dictionary_ptr_t			results_m;
		std::string					db_name_m;
		int							generation_m;
		
		db_list_t(std::size_t length, adobe::dictionary_t* results, const std::string& db_name, int generation)
		 : length_m(length), results_m(results), db_name_m(db_name), generation_m(generation) {}
		
		virtual ~db_list_t() {}
		virtual std::size_t size() const {
			return length_m;
		}
		virtual adobe::value_t at(unsigned int i) const {
			if( i >= length_m ) return adobe::value_t();
			return adobe::value_t(db_row_value_t(results_m, i, db_name_m, generation_m));
		}
		virtual bool empty() const {
			return length_m == 0;
		}
	};
	
	class db_query_t : public adobe::binder_t {
		util::db_ptr_t	db_m;		///< Our database.
		std::string	db_name_m;	///< Name of our database.
		std::string	query_m;	///< Query into database.
		adobe::binder_helper_t query_pump_m;	///< Query pump.
		adobe::binder_helper_t db_pump_m;	///< Db name pump.
		adobe::sheet_t::monitor_value_t value_monitor_m;
		adobe::sheet_t::monitor_active_t active_monitor_m;
		adobe::sheet_t&	sheet_m;
		bfs::path	scope_path_m;
		adobe::value_t	original_db_pump;
		adobe::value_t	original_query;
		boost::signals::scoped_connection db_connection_m;
		//
		/// Call our query and publish results. We are never active because
		/// we are always read-only.
		//
		void update() {
			static int generation = 0;
			if( !db_m || query_m.empty() ) {
				if( value_monitor_m ) value_monitor_m( adobe::value_t() );
				if( active_monitor_m ) active_monitor_m( false );
				return;
			}
			if( value_monitor_m ) {
				adobe::dictionary_t* results = new adobe::dictionary_t();
				std::size_t result_length;
				try {
					result_length = db_m->query( query_m, *results );
				} catch( std::exception& ex ) {
					result_length = 0;
					value_monitor_m( adobe::value_t() );
					if( active_monitor_m ) active_monitor_m( false );
					return;
				}
				value_monitor_m( adobe::value_t(
						util::concept::ListInstance( new db_list_t( result_length, results, db_name_m, generation++ ) ) ) );
			}
			if( active_monitor_m ) active_monitor_m( false );
		}
		void update_query( const adobe::value_t& query ) {
			std::string new_query = query.get<std::string>();
			if( query_m == new_query ) return;
			query_m = new_query;
			update();
		}
		void update_db( const adobe::value_t& db_value ) {
			std::string db_name = db_value.get<std::string>();
			if( db_name == db_name_m ) return;
			db_name_m = db_name;
			db_m = util::db::fromPath( scope_path_m / bfs::path( db_name_m, bfs::native ) );
			//
			// Make sure that we get notifications from the database.
			//
			db_connection_m = db_m->on_change( boost::bind( &db_query_t::update, this ) );
			update();
		}
	  public:
		db_query_t( const bfs::path& scope_path, const adobe::value_t& db_name, const adobe::value_t& query, adobe::sheet_t& sheet ) : query_pump_m( sheet ), db_pump_m( sheet ), sheet_m( sheet ), scope_path_m( scope_path ), original_db_pump( db_name ), original_query( query ) {
			query_pump_m.initialize( query, boost::bind( &db_query_t::update_query, this, _1 ), 0 );
			db_pump_m.initialize( db_name, boost::bind( &db_query_t::update_db, this, _1 ), 0 );
		}
		virtual ~db_query_t() {}
		virtual void set( const adobe::value_t& val ) {}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			active_monitor_m = monitor_active;
			update();
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			query_pump_m.substitute( name, val );
			db_pump_m.substitute( name, val );
			update();
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_query_t( scope_path_m, original_db_pump, original_query, sheet_m ) );
		}
	};
	adobe::value_t make_db_query_datapump( const bfs::path& scope_path, const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t db_name( parameters.at( adobe::static_name_t( "db" ) ) );
		adobe::value_t query( parameters.at( adobe::static_name_t( "query" ) ) );
		return adobe::value_t( adobe::binder_ptr_t( new db_query_t( scope_path, db_name, query, sheet ) ) );
	}
	//
	/// Used by db_row_t and db_value_t to generate the UPDATE statement by
	/// substituting new values in. Substitutes one value from the given
	/// dictionary into the string, returns the result. If not substitutions
	/// can be made (none remain in the given string) then the same string
	/// is returned.
	///
	/// \param	s	the string to substitute into.
	/// \param	d	the dictionary of strings to substitute from.
	/// \return	the substituted string, or s if no substitutions could be made.
	//
	static std::string substitute_dict( const std::string& s, const adobe::dictionary_t& d ) {
		std::string::size_type at = s.find( '@' );
		std::string::size_type semi = s.find( ';', at );
		if( at == std::string::npos ) return s;
		if( semi == std::string::npos ) return s;
		std::string token = s.substr( at + 1, semi - at - 1 );
		std::string replacement;
		adobe::name_t name( token.c_str() );
		if( d.count( name ) ) replacement = d.get<std::string>( name );
		std::string new_s( s );
		new_s.replace( at, semi - at + 1, replacement );
		return new_s;
	}
	class db_row_t : public adobe::binder_t {
		db_row_value_t	row_m;
		adobe::value_t	value_m;
		adobe::name_t	column_m;
		std::string		update_m;
		adobe::binder_helper_t	row_pump_m;
		adobe::sheet_t::monitor_active_t	active_monitor_m;
		adobe::sheet_t::monitor_value_t		value_monitor_m;
		bfs::path		scope_path_m;
		//
		// for making copies.
		// 
		adobe::sheet_t&		sheet_m;
		adobe::value_t		row_value_m;
		//
		/// Update the value of the row.
		///
		/// \param	val	the new row value.
		//
		void update_row( const adobe::value_t& val ) {
			row_m = val.get<db_row_value_t>();
			adobe::value_t new_val = row_m.get_column( column_m );
			if( new_val == value_m ) return;
			value_m = new_val;
			update();
		}
		//
		/// Update the value we're reflecting into the
		/// pipeline.
		//
		void update() {
			if( value_m != adobe::value_t() ) {
				if( active_monitor_m ) active_monitor_m( !update_m.empty() );
				if( value_monitor_m ) value_monitor_m( value_m );
				return;
			}
			if( active_monitor_m ) active_monitor_m( false );
			if( value_monitor_m ) value_monitor_m( adobe::value_t() );
		}
	  public:
		db_row_t( const bfs::path scope_path, const adobe::value_t& row, const adobe::name_t& column, const std::string& update, adobe::sheet_t& sheet )
		 : column_m( column ), update_m( update ), row_pump_m( sheet ), scope_path_m( scope_path ), sheet_m( sheet ), row_value_m( row ) {
			row_pump_m.initialize( row, boost::bind( &db_row_t::update_row, this, _1 ), 0 );
		}
		virtual ~db_row_t() {}
		virtual void set( const adobe::value_t& val ) {
			//
			// Walk through the update string, looking for "@" and ";".
			// 
			std::string update( update_m );
			std::string new_update( update_m );
			adobe::dictionary_t subst_dict = row_m.get_row_dictionary();
			if (val.type() == typeid(std::string))
				subst_dict.set( adobe::static_name_t( "NEWVAL" ), adobe::value_t( util::db::escape( val.get<std::string>() ) ) );
			else if (val.type() == typeid(double)) {
				std::stringstream ss;
				ss << val.get<double>();
				subst_dict.set( adobe::static_name_t( "NEWVAL" ), adobe::value_t( ss.str() ) );
			}
			do {
				update = new_update;
				new_update = substitute_dict( update, subst_dict );
			} while( update != new_update );
			//
			// Get the database name from the row.
			//
			util::db_ptr_t db = util::db::fromPath( scope_path_m / bfs::path( row_m.db_name_m, bfs::native ) );
			try {
				db->execute( update );
			} catch( std::exception& ex ) {
				//
				// XXX: remove debug code.
				//
				std::cerr << "ERROR: " << ex.what() << " RUNNING: " << update << std::endl;
			}
		}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			active_monitor_m = monitor_active;
			update();
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			row_pump_m.substitute( name, val );
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_row_t( scope_path_m, row_value_m, column_m, update_m, sheet_m ) );
		}
	};
	adobe::value_t make_db_row_datapump( const bfs::path& scope_path, const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t row( parameters.at( adobe::static_name_t( "row" ) ) );
		adobe::name_t column( parameters.get<std::string>( adobe::static_name_t( "column" ) ).c_str() );
		std::string update;
		if( parameters.count( adobe::static_name_t( "update" ) ) )
			update = parameters.get<std::string>( adobe::static_name_t( "update" ) );
		return adobe::value_t( adobe::binder_ptr_t( new db_row_t( scope_path, row, column, update, sheet ) ) );
	}
	//
	/// The db_column datapump extracts an entire column from a result set.
	//
	class db_column_t : public adobe::binder_t {
		adobe::binder_helper_t			result_set_pump_m;
		adobe::name_t					column_m;
		adobe::sheet_t&					sheet_m;
		adobe::value_t					results_m;

		adobe::sheet_t::monitor_value_t	value_monitor_m;
		adobe::value_t					last_m;

		void update( const adobe::value_t& val ) {
			//
			// REALLY need true concepts.
			//
			util::concept::List l = util::concept::get_list( val );
			if (l.size()) {
				try {
					db_row_value_t row = l.at(0).get<db_row_value_t>();
					last_m = row.get_entire_column( column_m );
				} catch( ... ) { last_m = adobe::value_t(); }
			} else last_m = adobe::value_t();
			if (value_monitor_m) value_monitor_m(last_m);
		}
	  public:
		db_column_t( const adobe::name_t& column, const adobe::value_t& results, adobe::sheet_t& sheet ) : result_set_pump_m(sheet), column_m(column), sheet_m(sheet), results_m(results) {
			result_set_pump_m.initialize( results, boost::bind( &db_column_t::update, this, _1 ), 0 );
		}
		virtual ~db_column_t() {}
		virtual void set( const adobe::value_t& ) {}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			if( monitor_active ) monitor_active( false );
			if( value_monitor_m ) value_monitor_m( last_m );
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			result_set_pump_m.substitute( name, val );
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_column_t( column_m, results_m, sheet_m ) );
		}
	};
	adobe::value_t make_db_column_datapump( const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t result_set(parameters.at(adobe::static_name_t("result_set")));
		adobe::name_t column(parameters.at(adobe::static_name_t("column")).get<std::string>().c_str());
		return adobe::value_t( adobe::binder_ptr_t( new db_column_t( column, result_set, sheet ) ) );
	}
	//
	/// The db_value datapump. Extracts a single value from the database
	/// and is able to update it.
	//
	class db_value_t : public adobe::binder_t {
		std::string				update_m;
		std::string				query_m;
		std::string				db_name_m;
		util::db_ptr_t				db_m;
		adobe::binder_helper_t			query_pump_m;
		adobe::binder_helper_t			db_pump_m;
		adobe::binder_helper_t			update_pump_m;
		adobe::sheet_t::monitor_active_t	active_monitor_m;
		adobe::sheet_t::monitor_value_t		value_monitor_m;
		boost::signals::scoped_connection	db_connection_m;
		bfs::path				scope_path_m;
		//
		// for making copies
		//
		adobe::sheet_t&				sheet_m;
		adobe::value_t				db_value_m;
		adobe::value_t				query_value_m;
		adobe::value_t				update_value_m;

		static void getResult( adobe::value_t& result, const adobe::dictionary_t& row ) {
			if( result != adobe::value_t() ) return;
			if( !row.size() ) return;
			result = row.begin()->second;
		}
		void update() {
			if( !db_m || query_m.empty() ) {
				if( active_monitor_m )	active_monitor_m( false );
				if( value_monitor_m )	value_monitor_m( adobe::value_t() );
				return;
			}
			if( value_monitor_m ) {
				//
				// Do the query. It may not succeed, if we're waiting on other
				// values to be set (for example), so swallow the exception in
				// that case.
				//
				adobe::value_t result;
				try {
					db_m->query( query_m, boost::bind( &db_value_t::getResult, boost::ref( result ), _1 ) );
				} catch( ... ) {
					value_monitor_m( adobe::value_t() );
					if( active_monitor_m ) active_monitor_m( false );
					return;
				}
				value_monitor_m( result );
			}
			if( active_monitor_m ) active_monitor_m( !update_m.empty() );
		}
		void update_db( const adobe::value_t& db ) {
			std::string db_name = db.get<std::string>();
			if( db_name == db_name_m ) return;
			db_name_m = db_name;
			db_m = util::db::fromPath( scope_path_m / bfs::path( db_name_m, bfs::native ) );
			//
			// Make sure that we get notifications from the database.
			//
			db_connection_m = db_m->on_change( boost::bind( &db_value_t::update, this ) );
			update();
		}
		void update_query( const adobe::value_t& query ) {
			std::string query_str = query.get<std::string>();
			if( query_m == query_str ) return;
			query_m = query_str;
			update();
		}
		void update_update( const adobe::value_t& update ) {
			bool was_empty( update_m.empty() );
			if( update.get( update_m ) ) {
				if( (was_empty != update_m.empty()) && active_monitor_m ) active_monitor_m( !update_m.empty() );
			} else {
				if( active_monitor_m ) active_monitor_m( false );
			}
		}
	  public:
		db_value_t( const bfs::path& scope_path, const adobe::value_t& db, const adobe::value_t& query, const adobe::value_t& update, adobe::sheet_t& sheet )
		 : query_pump_m( sheet ), db_pump_m( sheet ), update_pump_m( sheet ), scope_path_m( scope_path ), sheet_m( sheet ), db_value_m( db ),
		   query_value_m( query ), update_value_m( update )
		{
			db_pump_m.initialize( db, boost::bind( &db_value_t::update_db, this, _1 ), 0 );
			query_pump_m.initialize( query, boost::bind( &db_value_t::update_query, this, _1 ), 0 );
			update_pump_m.initialize( update, boost::bind( &db_value_t::update_update, this, _1 ), 0 );
		}
		virtual ~db_value_t() {}
		virtual void set( const adobe::value_t& val ) {
			if( !db_m ) return;
			//
			// Walk through the update string, looking for "@" and ";".
			// 
			std::string update( update_m );
			std::string new_update( update_m );
			adobe::dictionary_t subst_dict;
			subst_dict.set( adobe::static_name_t( "NEWVAL" ), adobe::value_t( util::db::escape( val.get<std::string>() ) ) );
			do {
				update = new_update;
				new_update = substitute_dict( update, subst_dict );
			} while( update != new_update );
			try {
				db_m->execute( update );
			} catch( std::exception& ex ) {
				//
				// XXX: remove debug code.
				//
				std::cerr << "ERROR: " << ex.what() << " RUNNING: " << update << std::endl;
			}
		}
		virtual void set_signals( const adobe::sheet_t::monitor_value_t& monitor_value, const adobe::sheet_t::monitor_active_t& monitor_active ) {
			value_monitor_m = monitor_value;
			active_monitor_m = monitor_active;
			update();
		}
		virtual void substitute( const adobe::name_t& name, const adobe::value_t& val ) {
			db_pump_m.substitute( name, val );
			query_pump_m.substitute( name, val );
			update_pump_m.substitute( name, val );
		}
		virtual void invoke() {}
		virtual adobe::binder_ptr_t copy() const {
			return adobe::binder_ptr_t( new db_value_t( scope_path_m, db_value_m, query_value_m, update_value_m, sheet_m ) );
		}
	};
	adobe::value_t make_db_value_datapump( const bfs::path& scope_path, const adobe::dictionary_t& parameters, adobe::sheet_t& sheet ) {
		adobe::value_t db( parameters.at( adobe::static_name_t( "db" ) ) );
		adobe::value_t query( parameters.at( adobe::static_name_t( "query" ) ) );
		adobe::value_t update;
		if( parameters.count( adobe::static_name_t( "update" ) ) )
			update = parameters.at( adobe::static_name_t( "update" ) );
		return adobe::value_t( adobe::binder_ptr_t( new db_value_t( scope_path, db, query, update, sheet ) ) );
	}
}
